package kafka_mq

import (
	"github.com/segmentio/kafka-go"
	"github.com/spf13/viper"
)

// NewKafkaConsumer 创建 Kafka 消费者
func NewKafkaConsumer(mqName string) *kafka.Reader {
	// 创建kafka reader
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   viper.GetStringSlice(mqName + ".brokers"), // Kafka 集群地址
		Topic:     viper.GetString(mqName + ".topic"),        // 要订阅的主题
		GroupID:   viper.GetString(mqName + ".group_id"),     // 消费组 ID（用于负载均衡和 Offset 管理）
		Partition: 0,                                         // 可省略，设置 GroupID 时 kafka-go 会自动处理分区
		MinBytes:  10e3,                                      // 10KB 最小获取消息体积
		MaxBytes:  10e6,                                      // 10MB 最大获取消息体积
	})

	return r
}

// NewKafkaProducer 创建一个 Kafka 生产者
func NewKafkaProducer(mqName string) *kafka.Writer {
	writer := kafka.NewWriter(kafka.WriterConfig{
		Brokers: viper.GetStringSlice(mqName + ".brokers"),
		Topic:   viper.GetString(mqName + ".topic"),
	})

	return writer
}
